package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSqlApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/3/14 09:22
 */
public class Dwd_04_DwdTradeCancelDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new Dwd_04_DwdTradeCancelDetail().init(
            3004,
            2,
            "Dwd_04_DwdTradeCancelDetail"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        // 下单表和取消表
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));
        
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_04_DwdTradeCancelDetail");
        // 2. 过虑 order_info 的数据: update   order_status: 1001->1003
        Table orderInfo = tEnv.sqlQuery("select " +
                                            "data['id'] id, " +
                                            "data['user_id'] user_id, " +
                                            "data['province_id'] province_id, " +
                                            "data['operate_time'] operate_time, " +  // 订单取消时间
                                            "ts " +
                                            "from ods_db " +
                                            "where `database`='gmall2022' " +
                                            "and `table`='order_info' " +
                                            "and `type`='update' " +
                                            "and `old`['order_status']='1001' " +  // 从未支付变成已取消
                                            "and `data`['order_status']='1003'");
        tEnv.createTemporaryView("order_info", orderInfo);
//        orderInfo.execute().print();
        
        // 3. 读取下单事务事实表
        tEnv.executeSql("create table dwd_trade_order_detail( " +
                            "id string, " +
                            "order_id string, " +
                            "user_id string, " +
                            "sku_id string, " +
                            "sku_name string, " +
                            "province_id string, " +
                            "activity_id string, " +
                            "activity_rule_id string, " +
                            "coupon_id string, " +
                            "date_id string, " +
                            "create_time string, " +
                            "source_id string, " +
                            "source_type string, " +
                            "source_type_name string, " +
                            "sku_num string, " +
                            "split_original_amount string, " +
                            "split_activity_amount string, " +
                            "split_coupon_amount string, " +
                            "split_total_amount string, " +
                            "ts bigint " +
                            ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "Dwd_04_DwdTradeCancelDetail"));
        
        
        // 4. join
        Table result = tEnv.sqlQuery("select " +
                                        "od.id, " +
                                        "od.order_id, " +
                                        "od.user_id, " +
                                        "od.sku_id, " +
                                        "od.sku_name, " +
                                        "od.province_id, " +
                                        "od.activity_id, " +
                                        "od.activity_rule_id, " +
                                        "od.coupon_id, " +
                                        "date_format(oi.operate_time, 'yyyy-MM-dd') operate_date_id, " +  // 取消的时候的年月日
                                        "oi.operate_time, " +
                                        "od.source_id, " +
                                        "od.source_type, " +
                                        "od.source_type_name, " +
                                        "od.sku_num, " +
                                        "od.split_original_amount, " +
                                        "od.split_activity_amount, " +
                                        "od.split_coupon_amount, " +
                                        "od.split_total_amount, " +
                                        "oi.ts " +
                                        "from " +
                                        "dwd_trade_order_detail od " +
                                        "join order_info oi " +
                                        "on od.order_id=oi.id ");
    
        // 5. 写出到 kafka 中
        tEnv.executeSql("create table dwd_trade_cancel_detail( " +
                                "id string, " +
                                "order_id string, " +
                                "user_id string, " +
                                "sku_id string, " +
                                "sku_name string, " +
                                "province_id string, " +
                                "activity_id string, " +
                                "activity_rule_id string, " +
                                "coupon_id string, " +
                                "date_id string, " +
                                "cancel_time string, " +
                                "source_id string, " +
                                "source_type_code string, " +
                                "source_type_name string, " +
                                "sku_num string, " +
                                "split_original_amount string, " +
                                "split_activity_amount string, " +
                                "split_coupon_amount string, " +
                                "split_total_amount string, " +
                                "ts bigint " +
                                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CANCEL_DETAIL));
    
        result.executeInsert("dwd_trade_cancel_detail");
        
    
    }
}
/*
交易域取消订单事务事实表
    粒度: 商品
    
        order_detail: insert
        
        order_info: update   order_status: 1001->1003
        
        order_detail_activity: insert
        
        order_detail_coupon: insert
        
        base_dic
        
    能否利用下单事务事实表的结果
    
        用下单事务事实表
            join
        order_info: update   order_status: 1001->1003

 */
